package liveData.migration.toPhoenix.sink;

import liveData.migration.toPhoenix.entity.FlatMessage;
import liveData.migration.toPhoenix.entity.JobSetting;
import liveData.migration.toPhoenix.util.ClassHelper;
import liveData.migration.toPhoenix.util.DmlType;
import liveData.migration.toPhoenix.util.SqlHelper;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.schema.MetaDataClient;
import org.apache.phoenix.schema.PTable;
import org.apache.phoenix.util.SchemaUtil;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.UUID;

/**
 * @program: dataStreming
 * @description:
 * @author: sunteng
 * @create: 2020-02-27 09:37
 **/
@Slf4j
public class CommonPhoenixSink extends RichSinkFunction<FlatMessage> {

    private Connection connection;
    private PreparedStatement upsertPreparedStatement;
    private PreparedStatement deletePreparedStatement;
    private ClassHelper classHelper;
    private SqlHelper sqlHelper;
    private PTable ptable;
    private JobSetting jobSetting;
    private String phoenixTable;
    private String mysqlPk;

    public CommonPhoenixSink(JobSetting jobSetting) {
        this.jobSetting = jobSetting;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        phoenixTable = StringUtils.upperCase(jobSetting.getSink().getPhoenixTable());
        mysqlPk = jobSetting.getSource().getMysqlPk();

        Class.forName(jobSetting.getSink().getPhoenixDriver());
        connection = DriverManager.getConnection(jobSetting.getSink().getPhoenixUrl());
        connection.setAutoCommit(true);

        sqlHelper = new SqlHelper();
        classHelper = new ClassHelper();

        ptable = getTableSchema(connection, phoenixTable);

        System.out.print("phoenixTable,ptable.getColumns().size()::" + phoenixTable + "," + ptable.getColumns().size());
        String upsertSql =
                "upsert into " + phoenixTable + " values(" + classHelper.makeSql(ptable.getColumns().size()) + ")";
        upsertPreparedStatement = connection.prepareStatement(upsertSql);

        String deleteSql = "delete from " + phoenixTable + " where " + "\"" + mysqlPk + "\"" + " = ?";
        deletePreparedStatement = connection.prepareStatement(deleteSql);

        super.open(parameters);
        log.debug("##open conn");
    }

    @Override
    public void invoke(FlatMessage flatMessage, Context context) throws Exception {

        if (flatMessage.getIsDdl()) {
            log.warn("the message is ddl [{}]", flatMessage);
        } else {

            Map<String, String> proMap = flatMessage.getData().get(0);

            log.debug("## is DML data ? [{}]", flatMessage.getIsDdl());
            if (flatMessage.getType().equals(DmlType.DELETE.toString())) {
                log.debug("##DELETE");
                deletePreparedStatement.setInt(1, Integer.parseInt(proMap.get(mysqlPk)));
                deletePreparedStatement.execute();

            } else if (flatMessage.getType().equals(DmlType.INSERT.toString()) || flatMessage.getType().equals(DmlType.UPDATE.toString())) {
                log.debug("##UPSERT");

                LinkedHashMap<String, Integer> columnTypes = getColumnSqlType(proMap);

                sqlHelper.setupStatement(columnTypes, upsertPreparedStatement);

                upsertPreparedStatement.executeUpdate();

            }
        }

        log.debug("##invoke");
    }

    private LinkedHashMap<String, Integer> getColumnSqlType(Map<String, String> proMap) {
        LinkedHashMap<String, Integer> columnSqlTypes = new LinkedHashMap<>();

        ptable.getColumns().forEach(pColumn -> {
            String key = proMap.get(pColumn.getName().getString());
            if (columnSqlTypes.containsKey(key)) {
                key = key + "," + UUID.randomUUID().toString();
            }
            columnSqlTypes.put(key, pColumn.getDataType().getSqlType());
        });

        return columnSqlTypes;
    }


    private PTable getTableSchema(Connection conn, String fullTableName) throws SQLException {
        PhoenixConnection hconn = conn.unwrap(PhoenixConnection.class);
        MetaDataClient mdc = new MetaDataClient(hconn);
        String schemaName = SchemaUtil.getSchemaNameFromFullName(fullTableName);
        String tableName = SchemaUtil.getTableNameFromFullName(fullTableName);
        return mdc.updateCache(schemaName, tableName).getTable();
    }

    @Override
    public void close() throws Exception {

        if (upsertPreparedStatement != null) {
            upsertPreparedStatement.close();
        }

        if (deletePreparedStatement != null) {
            deletePreparedStatement.close();
        }

        if (connection != null) {
            connection.close();
        }

        log.debug("##close conn");
        super.close();
    }
}
